package org.luxor.commons.zookeeper.client;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;
import org.luxor.commons.zookeeper.config.properties.ZookeeperProperties;
import org.luxor.commons.zookeeper.exception.ZKPathExistException;
import org.luxor.commons.zookeeper.exception.ZookeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

/**
 * Zookeeper Client
 *
 * @author Mr.Yan  @date: 2020/12/21
 */
public class AbstractZookeeperClient {
    private final Logger log = LoggerFactory.getLogger(this.getClass());

    protected CuratorFramework client;

    public AbstractZookeeperClient(CuratorFramework curatorFramework) {
        Assert.notNull(curatorFramework, "Spring容器中找不到[curatorFramework]的实例化对象!");
        this.client = curatorFramework;
    }

    /**
     * 查看客户端状态
     *
     * @return true 代表"连接正常"，false 代表"已关闭"或"未连接"
     */
    public boolean getState() {
        boolean isZkCuratorStarted = client.getState() == CuratorFrameworkState.STARTED;
        return isZkCuratorStarted;
    }

    /**
     * 节点已存在？
     *
     * @param path 节点路径
     * @return true 代表已存在，false代表不存在
     */
    public boolean checkExists(String path) throws ZookeeperException {
        try {
            Stat statExist = client.checkExists().forPath(path);
            return statExist != null;
        } catch (Exception e) {
            throw new ZookeeperException("检查节点是否已存在,失败.", e);
        }
    }

    /**
     * 创建节点
     *
     * @param path       节点路径
     * @param createMode 节点模式,默认为"永久模式"
     * @param data       节点数据
     * @param aclList    节点访问权限，默认为"开放访问"
     */
    public void create(String path, CreateMode createMode, byte[] data, ArrayList<ACL> aclList) throws ZookeeperException {
        try {
            client.create().creatingParentsIfNeeded() //如果父节点路径不存在，可以递归创建父节点
                    .withMode(createMode)
                    .withACL(aclList)
                    .forPath(path, data);
        } catch (KeeperException.NodeExistsException e) {
            throw new ZKPathExistException("创建新节点失败，[" + path + "]已存在.");
        } catch (Exception e) {
            throw new ZookeeperException("创建新节点失败.", e);
        }
    }

    /**
     * 创建持久节点
     *
     * @param path 节点路径
     * @param data 节点数据
     */
    public void create(String path, byte[] data) throws ZookeeperException {
        this.create(path, CreateMode.PERSISTENT, data, ZooDefs.Ids.OPEN_ACL_UNSAFE);
    }

    /**
     * 创建临时节点
     *
     * @param path 节点路径
     * @param data 节点数据
     */
    public void createEphemera(String path, byte[] data) throws ZookeeperException {
        this.create(path, CreateMode.EPHEMERAL_SEQUENTIAL, data, ZooDefs.Ids.OPEN_ACL_UNSAFE);
    }

    /**
     * 查看子节点列表
     *
     * @param path 父节点路径
     * @return 子节点列表
     */
    public List<String> getChildren(String path) throws ZookeeperException {
        try {
            List<String> children = client.getChildren().forPath(path);
            return children;
        } catch (KeeperException.NoNodeException e) {
            log.warn("获取数据失败. [" + path + "]不存在.");
            return Collections.emptyList();
        } catch (Exception e) {
            throw new ZookeeperException("查看子节点列表失败.", e);
        }
    }

    /**
     * 获取数据
     *
     * @param path 节点路径
     * @return 节点数据
     */
    public byte[] getData(String path) throws ZookeeperException {
        try {
            byte[] data = client.getData().forPath(path);
            return data;
        } catch (KeeperException.NoNodeException e) {
            log.warn("获取数据失败. [" + path + "]不存在.");
            return new byte[]{};
        } catch (Exception e) {
            throw new ZookeeperException("获取数据失败.", e);
        }
    }

    /**
     * 获取节点元数据
     *
     * @param path 节点路径
     * @return 节点元数据
     */
    public Stat getStat(String path) throws ZookeeperException {
        Stat stat = new Stat();
        try {
            client.getData().storingStatIn(stat).forPath(path);
            return stat;
        } catch (KeeperException.NoNodeException e) {
            log.warn("获取数据失败. [" + path + "]不存在.");
            return stat;
        } catch (Exception e) {
            throw new ZookeeperException("获取数据失败.", e);
        }
    }

    /**
     * 修改数据
     *
     * @param path 节点路径
     * @param data 节点数据
     */
    public void setData(String path, byte[] data) throws ZookeeperException {
        try {
            if (!checkExists(path)) {
                client.create().creatingParentsIfNeeded().forPath(path, data);
            }
            client.setData().forPath(path, data);
        } catch (Exception e) {
            throw new ZookeeperException("修改数据失败.", e);
        }
    }

    /**
     * 删除节点
     *
     * @param path 节点路径
     */
    public void delete(String path) throws ZookeeperException {
        try {
            if (ZookeeperProperties.ROOT_PATH.equalsIgnoreCase(path)) {
                log.warn("Cannot delete Root('/') nodes");
                return;
            }
            client.delete().forPath(path);
        } catch (KeeperException.NoNodeException e) {
            log.warn("删除节点失败. [" + path + "]不存在.");
            return;
        } catch (Exception e) {
            throw new ZookeeperException("删除节点失败.", e);
        }
    }

    /**
     * 删除节点及其全部子节点
     *
     * @param path 节点路径
     */
    public void deleteAll(String path) throws ZookeeperException {
        try {
            if (ZookeeperProperties.ROOT_PATH.equalsIgnoreCase(path)) {
                log.warn("Cannot delete Root('/') nodes");
                return;
            }
            client.delete().deletingChildrenIfNeeded()//如果存在子子节点路径，可以强制递归删除
                    .forPath(path);
        } catch (KeeperException.NoNodeException e) {
            log.warn("删除节点及其全部子节点失败. [" + path + "]不存在.");
            return;
        } catch (Exception e) {
            throw new ZookeeperException("删除节点及其全部子节点失败.", e);
        }
    }

}
